package com.yh.service.impl;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yh.config.OtherSource;
import com.yh.dao.*;
import com.yh.entity.*;
import com.yh.pojo.MqMessage;
import com.yh.service.MovieService;
import com.yh.service.redis.UpdateDataToRedis;
import com.yh.utils.Constants;
import com.yh.utils.ImgUrl;
import com.yh.vo.MovieRoleVO;
import com.yh.vo.MovieVO;
import io.minio.MinioClient;
import org.apache.dubbo.config.annotation.DubboService;
import org.apache.log4j.Logger;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.util.ObjectUtils;

import java.util.*;

/**
 * @author yuhuan
 * @date 2022/01/13
 */
@DubboService
public class MovieServiceImpl implements MovieService {
    /**
     * 日志
     */
    private static final Logger logger = Logger.getLogger(MovieServiceImpl.class);

    @Autowired
    private MovieDao movieDao;
    @Autowired
    private MovieTypeDao typeDao;
    @Autowired
    private MovieCountryDao countryDao;
    @Autowired
    private DirectorDao directorDao;
    @Autowired
    private CommentDao commentDao;
    @Autowired
    private ActorDao actorDao;
    @Autowired
    private LanguageVersionDao versionDao;
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private MinioClient minioClient;
    @Autowired
    private OtherSource otherSource;
    @Autowired
    private UpdateDataToRedis redis;
    @Autowired
    private Jackson2JsonMessageConverter jackson2JsonMessageConverter;

    @Override
    public List<Movie> findMovieAll() {
        List<Movie> movies = movieDao.findMovieAll();
        return ImgUrl.movieImageHandler(minioClient, otherSource, movies);
    }

    @Override
    public Movie findMovieById(int movieId) {
        // 判断影片在缓存中是否存在
        Boolean hasKey = redisTemplate.hasKey(redis.createMovieIdInRedis(movieId));
        if (hasKey) {
            MovieVO movieVO = findMovieDetailByIdFromRedis(movieId);
            return movieVO.getMovie();
        }
        Movie movie = movieDao.findMovieById(movieId);
        if (ObjectUtils.isEmpty(movie)) {
            return movie;
        }
        return ImgUrl.movieImageHandler(minioClient, otherSource, movie);
    }

    @Override
    public Movie findMovieByIdNeedIgnoreState(int movieId) {
        Movie movie = findMovieById(movieId);
        if (ObjectUtils.isEmpty(movie)) {
            movie = movieDao.findMovieByIdNeedIgnoreState(movieId);
            return ImgUrl.movieImageHandler(minioClient, otherSource, movie);
        }
        return movie;
    }

    @Override
    public MovieVO findMovieDetailById(int movieId) {
        // 找到电影实体信息并根据电影信息得到初始的电影响应模型（只含有电影的基本信息和角色关系映射列表）
        Movie movie = movieDao.findMovieById(movieId);
        // 如果查询结果为空，返回一个没有数据的电影模型
        if (ObjectUtils.isEmpty(movie)) {
            return null;
        }
        // 处理图片地址
        movie = ImgUrl.movieImageHandler(minioClient, otherSource, movie);
        MovieVO primaryMovieVO = new MovieVO(movie);
        // 初始化影片角色模型得到半完整的电影响应模型（包含演员的信息）
        MovieVO halfMovieVO = initActorRoleVo(primaryMovieVO);
        // 找到导演实体信息并封装到电影模型
        Director director = directorDao.findDirectorById(movie.getMovieDirectorId());
        // 处理图片地址
        director = ImgUrl.directorImageHandler(minioClient, otherSource, director);
        halfMovieVO.setDirector(director);
        return halfMovieVO;
    }

    @Override
    public MovieVO findMovieDetailByIdFromRedis(int movieId) {
        // 获取字符串类型的操作模板
        ValueOperations operations = redisTemplate.opsForValue();
        // 拼接该影片在缓存中的ID
        String idInRedis = redis.createMovieIdInRedis(movieId);
        // 到redis中获取电影的缓存信息
        String movieVoJson = (String) operations.get(idInRedis);
        // 如果缓存中没有找到该电影的信息，就到数据库查找并把查询结果缓存到redis
        if (ObjectUtils.isEmpty(movieVoJson)) {
            MovieVO movieVO = findMovieDetailById(movieId);
            if (!ObjectUtils.isEmpty(movieVO)) {
                redis.saveMovieDetailToRedis(movieVO);
            }
            return movieVO;
        }
        try {
            // 从缓存中取出数据并序列化
            MovieVO movieVO = objectMapper.readValue(movieVoJson, MovieVO.class);
            return movieVO;
        } catch (JsonProcessingException e) {
            return null;
        }
    }

    @Override
    public MovieVO findMovieByIdNeedStarName(int movieId) {
        // 判断影片在缓存中是否存在
        Boolean hasKey = redisTemplate.hasKey(redis.createMovieIdInRedis(movieId));
        if (hasKey) {
            MovieVO movieVO = findMovieDetailByIdFromRedis(movieId);
            return splicStarNameToMovieVO(movieVO);
        }
        // 到数据库找到影片信息
        Movie movie = movieDao.findMovieById(movieId);
        if (ObjectUtils.isEmpty(movie)) {
            return null;
        }
        movie = ImgUrl.movieImageHandler(minioClient, otherSource, movie);
        MovieVO movieVO = new MovieVO(movie);
        // 根据影片找到并拼接主演的姓名；如果没有设置主演信息，将展示默认的消息
        List<String> movieStar = movie.getMovieStar();
        List<String> starNames = new ArrayList<>();
        if (ObjectUtils.isEmpty(movieStar)) {
            // 设置默认的显示内容
            starNames.add(Constants.MovieData.NULL_MOVIE_STAR);
        }else {
            // 处理主演的姓名
            movieStar.forEach(s -> {
                Actor actor = actorDao.findActorById(Integer.valueOf(s));
                starNames.add(actor.getActorName());
            });
        }
        movieVO.setStarNameToStr(listSplicToStr(starNames));
        return movieVO;
    }

    @Override
    public Boolean moviePreheatToRedis() {
        List<Movie> movies = findMovieAll();
        if (ObjectUtils.isEmpty(movies)) {
            return false;
        }
        // 将影片信息发布到消息队列
        for (Movie movie : movies) {
            saveDataToRedisAndEs(movie, true);
        }
        return true;
    }

    @Override
    public int addMovie(MovieVO movieVO) {
        Movie movie = new Movie();
        BeanUtils.copyProperties(movieVO, movie);
        movieDao.insertMovie(movie);
        // 将影片信息添加到搜索引擎
        saveDataToRedisAndEs(movie, false);
        return Constants.ResponseDataState.RESP_SUCCESS;
    }

    @Override
    public int updateMovie(MovieVO movieVO) {
        Movie movie = new Movie();
        BeanUtils.copyProperties(movieVO, movie);
        movieDao.updateMovie(movie);
        // 清理缓存中的影片数据
        boolean success = redis.clearMovie(movieVO.getMovieId());
        if (success) {
            // 重新建立缓存和搜索引擎中的搜索文档。为了保证在用户没有修改海报时；也可以正确处理海报地址，这里选择仅发布ID
            saveDataToRedisAndEs(movie, true);
        }
        return Constants.ResponseDataState.RESP_SUCCESS;
    }

    @Override
    public int deleteMovie(int movieId) {
        // 删除数据库中的数据
        movieDao.removeMovie(movieId,Constants.INERT_DATA);
        // 清理缓存中的数据
        redis.clearMovie(movieId);
        // 清理搜索引擎中的数据
        MqMessage message = new MqMessage();
        message.setType(Constants.MqMessageData.DEL_MESSAGE);
        message.setData(movieId);
        pushMessageToMq(message);
        return Constants.ResponseDataState.RESP_SUCCESS;
    }

    @Override
    public List<LanguageVersion> findLanguageVersionAll() {
        return versionDao.findVersionAll();
    }

    @Override
    public List<MovieType> findMovieTypeAll() {
        return typeDao.findTypeAll();
    }

    @Override
    public int addMovieType(MovieType type) {
        return typeDao.insertType(type);
    }

    @Override
    public List<MovieCountry> findMovieCountryAll() {
        return countryDao.findCountryAll();
    }

    @Override
    public int addMovieCountry(MovieCountry country) {
        return countryDao.insertCountry(country);
    }

    /**
     * 保存数据到缓存和搜索引擎；缓存只保存热点影片（所以会对影片进行过滤）
     * 该方法会将影片的ID或整个影片的信息发布到队列，通过使用'useId'来决定是否发布影片的ID到队列。
     * 如果设置为false，则表示放弃使用ID作为消息；而是将整个影片发布到消息队列，这样消费端就不需要
     * 再次执行查询了，但这样会消耗大量的内存空间。
     */
    private void saveDataToRedisAndEs(Movie movie, boolean useId) {
        Double movieScore = movie.getMovieScore();
        // 筛选高分影片
        if (!ObjectUtils.isEmpty(movieScore) && movieScore >= Constants.MovieData.HIGH_SCORE_VIDEOS) {
            // 查询影片详情并保存到 redis
            findMovieDetailByIdFromRedis(movie.getMovieId());
        }
        // 设置发送消息的消息转换器
        rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter);
        // 发布的消息将由接收字符串的消费者处理；注意不要直接发布int类型的数据
        if (useId) {
            rabbitTemplate.convertAndSend(
                    Constants.RabbitMqData.PREHEAT_EXCHANGE_NAME,
                    Constants.RabbitMqData.PREHEAT_BINDING_KEY,
                    String.valueOf(movie.getMovieId()));
            return;
        }
        // 发布的消息将由接收影片实体的消费者处理
        rabbitTemplate.convertAndSend(
                Constants.RabbitMqData.PREHEAT_EXCHANGE_NAME,
                Constants.RabbitMqData.PREHEAT_BINDING_KEY,
                movie);
    }

    /**
     * 发布普通消息到队列（统一消息模型）
     */
    private void pushMessageToMq(MqMessage message) {
        // 设置发送消息的消息转换器
        rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter);
        rabbitTemplate.convertAndSend(
                Constants.RabbitMqData.PREHEAT_EXCHANGE_NAME,
                Constants.RabbitMqData.PREHEAT_BINDING_KEY,
                message);
    }

    /**
     * 将主演的姓名拼接成字符串
     */
    private MovieVO splicStarNameToMovieVO(MovieVO movieVO) {
        List<MovieRoleVO> actorRoles = movieVO.getActorRoles();
        if (ObjectUtils.isEmpty(actorRoles)) {
            return movieVO;
        }
        Set<String> starNames = new LinkedHashSet<>();
        // 判断角色是否为主演饰演，如果是就将演员的姓名丢到set集合中
        actorRoles.forEach(role -> {
            if (role.getIsStar()) {
                starNames.add(role.getActor().getActorName());
            }
        });
        // 以空格拼接主演的姓名
        movieVO.setStarNameToStr(listSplicToStr(starNames));
        return movieVO;
    }

    /**
     * 用空格将集合中的元素拼接成字符串
     */
    private String listSplicToStr(Collection<String> starNames) {
        return String.join(" ", starNames);
    }

    /**
     * 初始化影片角色模型（根据映射关系将演员信息绑定到正确的电影角色模型中）
     */
    private MovieVO initActorRoleVo(MovieVO primaryMovieVO) {
        // 取出演员与角色的关系映射表和角色模型列表
        List<String> actorNames = primaryMovieVO.getActorNames();
        List<MovieRoleVO> movieRoleVos = primaryMovieVO.getActorRoles();
        // 根据演员的ID找到所有的演员信息
        List<Actor> actors = initActorList(primaryMovieVO.getMovie().getMovieActorIds());
        // 如果映射关系为空，就将电影模型的数据状态标记为不完整的数据
        if (ObjectUtils.isEmpty(actorNames) || ObjectUtils.isEmpty(movieRoleVos)) {
            logger.debug(primaryMovieVO.getMovie().getMovieId() + "号影片的演员映射关系不存在");
            return primaryMovieVO;
        }
        if (ObjectUtils.isEmpty(actors)) {
            logger.debug(primaryMovieVO.getMovie().getMovieId() + "号影片的演员信息没有找到");
            return primaryMovieVO;
        }
        // 根据映射关系将演员信息绑定到正确的角色模型上
        for (int i = 0; i < actorNames.size(); i++) {
            for (Actor actor : actors) {
                if (actor.getActorName().equals(actorNames.get(i))) {
                    movieRoleVos.get(i).setActor(actor);
                    // 标记该演员是否为主演
                    List<String> movieStar = primaryMovieVO.getMovie().getMovieStar();
                    if (movieStar.contains(String.valueOf(actor.getActorId()))) {
                        movieRoleVos.get(i).setIsStar(true);
                    } else {
                        movieRoleVos.get(i).setIsStar(false);
                    }
                    break;
                }
            }
        }
        return primaryMovieVO;
    }

    /**
     * 根据ID集初始化所有的演员信息
     */
    private List<Actor> initActorList(List<String> actorIds) {
        List<Actor> actors = new ArrayList<>();
        if (ObjectUtils.isEmpty(actorIds)) {
            return actors;
        }
        actorIds.forEach(actorId -> {
            Actor actor = actorDao.findActorById(Integer.valueOf(actorId));
            // 如果查询结果不为空才添加到列表中，防止演员信息绑定到角色模型时出现空指针
            if (!ObjectUtils.isEmpty(actor)) {
                // 处理图片地址
                actor = ImgUrl.actorImageHandler(minioClient, otherSource, actor);
                actors.add(actor);
            }
        });
        return actors;
    }
}
